package com.bb.blog.code.threading.example;

import org.apache.zookeeper.util.CircularBlockingQueue;

import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 简易版线程池
 */
public class ExecuteServiceExample extends AbstractExecutorService {
    private int coreThreadSize;
    private int maxThreadSize;
    private AtomicInteger workCount;
    // runState is stored in the high-order bits
    private AtomicInteger runState = new AtomicInteger(-1);


    private static final int RUNNING = -1;
    private static final int SHUTDOWN = 0;
    private static final int TERMINATED = 3;

    /**
     * 工作队列
     */
    private final BlockingQueue<Runnable> workQueue;

    private ReentrantLock mainLock = new ReentrantLock();

    private final HashSet<Worker> workerHashSet = new HashSet<Worker>();

    private AtomicLong completedWorkCount = new AtomicLong(0);

    private volatile ThreadFactory threadFactory;

    private volatile long keepAliveTime;


    public ExecuteServiceExample(int queueSize, int coreSize, int maxSize, ThreadFactory threadFactory) {
        this.coreThreadSize = coreSize;
        this.maxThreadSize = maxSize;
        this.workCount = new AtomicInteger(0);
        this.workQueue = new ArrayBlockingQueue<>(queueSize);
        this.threadFactory = threadFactory;
        this.keepAliveTime = 60;

    }


    @Override
    public void shutdown() {

    }

    @Override
    public List<Runnable> shutdownNow() {
        return null;
    }

    @Override
    public boolean isShutdown() {
        return false;
    }

    @Override
    public boolean isTerminated() {
        return false;
    }

    public boolean isRunning() {
        return runState.get() == RUNNING;
    }


    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void execute(Runnable command) {
        if (command == null) {
            return;
        }
        if (isRunning()) {
            if (!(workCount.get() >= coreThreadSize && workQueue.offer(command))) {
                addWorker(command, false);
            } else if (workCount.get() < coreThreadSize) {
                addWorker(command, true);

            }
        } else {
            throw new RuntimeException("状态不对劲");
        }
    }

    private boolean addWorker(Runnable command, boolean core) {
        for(;;){
            if(!isRunning()){
                return false;
            }
            int wc = workCount.get();
            if(wc > (core?coreThreadSize:maxThreadSize)){
                return false;
            }
            if(workCount.compareAndSet(wc,wc+1)){
                break;
            }
        }
        boolean workerStarted = false;
        boolean workAdded = false;
        Worker worker = null;
        try{
            worker = new Worker(command);
            final Thread t = worker.thread;
            if(t!= null){
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try{
                    if(isRunning()){
                        workerHashSet.add(worker);
                        workAdded = true;
                    }else{
                        throw new RuntimeException("状态不对劲");
                    }
                }
                finally {
                    mainLock.unlock();
                }

            }
            if(workAdded){
                t.start();
                workerStarted = true;
            }

        }
        finally {
            if(!workAdded){
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                workerHashSet.remove(worker);
                workCount.decrementAndGet();
                tryTerminate();
                mainLock.unlock();
            }

        }
        return workerStarted;
    }

    private void tryTerminate() {
        if(isTerminated()){
            System.out.println("线程池裂开了");
        }
    }

    public ThreadFactory getThreadFactory() {
        return threadFactory;
    }


    private void runWorker(Worker worker) {
        // worker 正在跑的线程
        Thread t = Thread.currentThread();
        // 正在执行的task
        Runnable task = worker.firstTask;
        worker.firstTask = null;
        worker.unlock();
        boolean completedAbruptly = true;
        try {
            while (task != null || (task = takeTask()) != null) {
                worker.lock();
                try {
                    System.out.println("worker: " + worker + "开始执行任务：" + task.hashCode() + " at" + System.currentTimeMillis());
                    task.run();
                    System.out.println("worker: " + worker + "完成任务：" + task.hashCode() + " at" + System.currentTimeMillis());
                } finally {
                    task = null;
                    worker.completedTasks++;
                    worker.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkExit(worker, completedAbruptly);
        }


    }

    private Runnable takeTask() {
        boolean timeOut = false;
        for(;;){
            if(isTerminated()){
                throw new RuntimeException("线程池裂开了");
            }
            int wc = workCount.get();
            boolean timed = wc> coreThreadSize;

            if(timed && timeOut){
                return null;
            }
            try{
                Runnable r = timed?workQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS):
                        workQueue.take();
                if(r!=null){
                    return r;
                }
                timeOut= true;
            } catch (InterruptedException e) {
                timeOut = false;
            }
        }

    }

    private void processWorkExit(Worker worker, boolean completedAbruptly) {
        mainLock.lock();
        long count = completedWorkCount.get();
        completedWorkCount.compareAndSet(count, count + worker.completedTasks) ;
        workerHashSet.remove(worker);
        workCount.decrementAndGet();
        mainLock.unlock();

        tryTerminate();
    }

    class Worker extends AbstractQueuedSynchronizer implements Runnable {
        /**
         * 当前Worker正在执行的线程。
         */
        final Thread thread;

        Runnable firstTask;

        volatile long completedTasks;
        private final String name;

        Worker(Runnable firstTask) {
            setState(-1);
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
            name = thread.getName();
        }

        @Override
        public void run() {
            runWorker(this);
        }

        // =====================  aqs 重写的方法
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }

        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        public void lock() {
            acquire(1);
        }

        public boolean tryLock() {
            return tryAcquire(1);
        }

        public void unlock() {
            release(1);
        }

        public boolean isLocked() {
            return isHeldExclusively();
        }

        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

    public static void main(String[] args) {
//        ExecuteServiceExample example = new ExecuteServiceExample(100, 2, 10, Executors.defaultThreadFactory());
        ExecutorService example = new ThreadPoolExecutor(3, 6, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<>(30),
                Executors.defaultThreadFactory(), new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.out.println("拒绝了");
            }
        });
        for (int i = 0; i < 100; i++) {
            int k = i;
            example.execute(() -> {
                System.out.println("fuck you " + k + " times");
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
//        try {
//            Thread.sleep(10000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        System.out.println(example.completedWorkCount.get());


    }
}
